package org.jxkj.app

import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.jxkj.data.IDaoProvider
import org.jxkj.util.JzCondition

/**
 * 1、母线电压计算
 */
object BusVoltageQualification {
  /**
   * 母线电压计算
   *
   * @param ss
   * @param sc
   * @param hTable
   * @param time
   * @param step
   */
  def calcBusVoltage(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {
    var startTime: String = time(0) //+" 00:00:00"
    val firstTime = time(0)
    val endTime: String = time(1) //hu+" 23:59:59"
    //读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_eo_bus_volconf") //大江二江配置表
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    import ss.implicits._
    //    jdbcDF.show()
    //val workingDF = jdbcDF.select("project_name","stat_quarter","stat_beg_time","stat_end_time","bus_voltage_upper","bus_voltage_lower","bus_voltage_deactive").dropDuplicates("project_name","stat_quarter","stat_quarter","stat_beg_time")
    jdbcDF.createOrReplaceTempView("hms_bus_voltage_scope_conf")

    ss.sql("select project_name,case when substr(stat_quarter,6,1)=1 then concat(substr(stat_quarter,1,4),'/01/01 ')" +
      "when substr(stat_quarter,6,1)=2 then concat(substr(stat_quarter,1,4),'/04/01 ')" +
      "when substr(stat_quarter,6,1)=3 then concat(substr(stat_quarter,1,4),'/07/01 ')" +
      "when substr(stat_quarter,6,1)=4 then concat(substr(stat_quarter,1,4),'/10/01 ') end as beg_date," +
      "case when substr(stat_quarter,6,1)=1 then concat(substr(stat_quarter,1,4),'/03/31 ')" +
      "when substr(stat_quarter,6,1)=2 then concat(substr(stat_quarter,1,4),'/06/30 ')" +
      "when substr(stat_quarter,6,1)=3 then concat(substr(stat_quarter,1,4),'/09/30 ')" +
      "when substr(stat_quarter,6,1)=4 then concat(substr(stat_quarter,1,4),'/12/31 ') end as end_date," +
      "lpad(stat_beg_time,8,'0') as beg_hhmmss,lpad(stat_end_time,8,'0') as end_hhmmss,bus_voltage_upper,bus_voltage_lower,bus_voltage_deactive from hms_bus_voltage_scope_conf")
      .createOrReplaceTempView("hms_bus_voltage_scope_conf2")
    //     ss.sql("select * from hms_bus_voltage_scope_conf2").show(54)

    val jdbc2DF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url)
      .option("dbtable", "hms_bus_voltage_point_conf") //edos库点位配置表
      .option("user", MysqlDml.user)
      .option("password", MysqlDml.password)
      .option("driver", MysqlDml.driver2)
      .load()
    jdbc2DF.createOrReplaceTempView("hms_bus_voltage_point_conf")
    //    jdbc2DF.show()
    //.filter($"unit_name"==="07F")
    val arrayString: Array[String] = jdbc2DF.filter($"bus_id".substr(2, 1) === "母").map { x => x.getString(1) }.collect()
    //val arrayString = Array("JKXT_DJ_AI03825","JKXT_DJ_AI03871")
    println(arrayString.mkString("---"))


    GetData.getBySecond(ss, sc, hTable, arrayString, time, step, 60)
    ss.catalog.cacheTable("t_ods_hbase_edos_tmp")
    ss.sql("select * from t_ods_hbase_edos_tmp ").show()
    //ss.sql("select *from t_ods_hbase_edos_tmp where checkPoint='CM2212AMEAS_Bus1MMXU1_PPV_phsAB_cVal_mag_f' order by checkPoint,createTime").show(200)

    /*20210122加
    * 大江二江越线 异常停电判断
    * 大江 ：异常<0  停电<500< 越下限 <527<= 正常<=537<越上限
    *二江       0         200         224          231
    * 2021-03-29 修改：b.pvalue>=c.bus_voltage_deactive 原因是因为当值为零的时候给过滤了所以要加等于号
    *           修改：substr(c.beg_date,1,10) and substr(c.end_date,1,10) 这里是因为跨年时间问题的修改
    * 2021-3-30 修改a.project_name = '大江' and b.pvalue <= c.bus_voltage_upper and b.pvalue >= c.bus_voltage_lower 原本是 <=527 >=537
    * */
    // 1:越线
    val resultDf1 = ss.sql("select a.project_name,bus_id,b.createTime as stat_time,c.bus_voltage_upper,c.bus_voltage_lower,round(b.pvalue,2) bus_voltage," +
      //      "case when b.pvalue between bus_voltage_lower and bus_voltage_upper then 0 else 1 end bus_voltage_status," +
      "case when a.project_name = '大江' and b.pvalue <= c.bus_voltage_upper and b.pvalue >= c.bus_voltage_lower  then 0 	when a.project_name = '二江' and b.pvalue >= c.bus_voltage_lower and b.pvalue <= c.bus_voltage_upper  then 0  " +
      "when a.project_name = '大江' and b.pvalue < 500 and b.pvalue >= 0  then 3   when a.project_name = '二江' and b.pvalue >= 0 and b.pvalue < 220  then 3  " +
      "when '大江' = a.project_name  and b.pvalue > c.bus_voltage_upper  then 1    when '二江' = a.project_name  and b.pvalue > c.bus_voltage_lower  then 1 	" +
      "when a.project_name = '大江' and b.pvalue < c.bus_voltage_lower and b.pvalue >= 500  then 2 	when a.project_name = '二江' and b.pvalue >= 200 and b.pvalue < c.bus_voltage_lower  then 2  " +
      " when a.project_name = '大江' and b.pvalue < 0 and b.pvalue is NULL then 4  when a.project_name = '二江' and b.pvalue < 0 and b.pvalue is NULL then 4 end bus_voltage_status," +

      "'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time from hms_bus_voltage_point_conf a" +
      "  join t_ods_hbase_edos_tmp b on a.point_code=b.checkPoint " +
      " left join hms_bus_voltage_scope_conf2 c on a.project_name=c.project_name and substr(b.createTime,1,10) between substr(c.beg_date,1,10) and substr(c.end_date,1,10) and  substr(b.createTime,12,8)>= c.beg_hhmmss and substr(b.createTime,12,8)<=c.end_hhmmss" +
      " where b.pvalue >= c.bus_voltage_deactive" +
      " and substr(a.bus_id,-1)='母' ")
    //resultDf1.filter("bus_id='三母'").show()
    resultDf1.createOrReplaceTempView("t_etl")
    //resultDf1.show()

    val resultDf2 = ss.sql("select project_name,'' bus_id,stat_time,0 bus_voltage_upper,0 bus_voltage_lower,0 bus_voltage,max(bus_voltage_status) as bus_voltage_status," +
      "'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time" +
      " from t_etl a group by project_name,stat_time")

    val resultDf3 = ss.sql("select '全厂' project_name,'' bus_id,stat_time,0 bus_voltage_upper,0 bus_voltage_lower,0 bus_voltage,max(bus_voltage_status) as bus_voltage_status," +
      "'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time" +
      " from t_etl a" +
      " group by stat_time")

    val resultDf = resultDf1.union(resultDf2).union(resultDf3)
    resultDf.createOrReplaceTempView("t_etl2")
    //resultDf.show()

    //测试2021暂时关闭
    //    /*
    val tableName = "hms_eo_bus_volpassrate"

    //    resultDf.createOrReplaceTempView("t_resultDf")
    //    resultDf.show()
    //    MysqlDml.delete2(s"delete from $tableName where stat_time between  '$startTime'  and  '$endTime'")
    MysqlDml.delete2(s"delete from $tableName where stat_time between  '$startTime'  and  '$endTime' ")

    //写mysql
    resultDf.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tableName)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()
    //    ss.catalog.uncacheTable("t_ods_hbase_edos_tmp")

    //*///测试2021 所以先暂时注释
    //////////////测试2021/////
    //    2021-01-18+ 按小时计算故需将0点到开始时间的数据+结束时间到23:59:59取出来一起汇总成天
    val t1 = startTime.substring(0, 10) + " 00:00:00"
    val t2 = endTime.substring(0, 10) + " 23:59:59"

    /** val resultDf5Before=ss.read
     * .format("jdbc")
     * .option("url",MysqlDml.url2)
     * .option("dbtable","hms_eo_bus_volpassrate")//
     * .option("user",MysqlDml.user2)
     * .option("password",MysqlDml.password2)
     * .option("driver",MysqlDml.driver2)
     * .load()
     *
     *
     * val temp1= resultDf5Before.filter("stat_time>='"+t1+"'") .filter("stat_time<'"+startTime+"'")//0点到开始
     * val temp2= resultDf5Before.filter("stat_time>'"+endTime+"'").filter("stat_time<='"+t2+"'")//结束到24点
     * val aa=temp1.union(temp2)
     * aa.createOrReplaceTempView("t_etl2Before")//    aa.show()
     * val rtemp=ss.sql(" select project_name, bus_id,stat_time, bus_voltage_upper, bus_voltage_lower, bus_voltage, bus_voltage_status, " +
     * "project_id, org_id, sub_org_id, create_person, create_time, modify_person, modify_time from t_etl2Before ")
     * val resultDftemp=rtemp.union(resultDf)
     * resultDftemp.createOrReplaceTempView("t_etl2Total")
     *
     * /////按天汇总
     * /*
     * 2021-02-25：添加 bus_voltage_status(当前电压状态) voltage_max(本月最大电压值) voltage_min(本月最小电压值) 的逻辑判断
     * */
     * val resultDf5 = ss.sql("select project_name,bus_id,substr(stat_time,1,10) as stat_time," +
     * "sum( case when project_name = '大江' and bus_voltage <= 537 and bus_voltage >= 527  then 1 when project_name = '二江' and bus_voltage >= 224 and bus_voltage <= 231  then 1 else 0 end  ) bus_voltage_cnt," +
     * "sum( case when project_name = '大江' and bus_voltage < 500 and bus_voltage >= 0  then 1 when project_name = '二江' and bus_voltage >= 200 and bus_voltage < 220  then 1 else 0 end  ) power_outage," +
     * "sum( case  when '大江' = project_name  and bus_voltage > 537  then 1 when '二江' = project_name  and bus_voltage > 231  then 1 else 0 end ) interval_upper," +
     * "sum( case when project_name = '大江' and bus_voltage < 527 and bus_voltage >= 500  then 1 when project_name = '二江' and bus_voltage >= 200 and bus_voltage < 224  then 1 else 0 end  ) interval_lower," +
     * "sum( case  when project_name = '大江' and bus_voltage < 0 and bus_voltage is NULL then 1 when project_name = '二江' and bus_voltage < 0 and bus_voltage is NULL then 1 else 0 end  ) data_exception," +
     * "( case when avg(bus_voltage_status) > 0 then 1 else 0 end ) bus_voltage_status," +
     * "max( bus_voltage ) voltage_max," +
     * "min( bus_voltage ) voltage_min," +
     * " 'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time " +
     * "from t_etl2Total a  group by project_name,bus_id,substr(stat_time,1,10)"
     * //    "from t_etl2 a  group by project_name,bus_id,substr(stat_time,1,10)"
     * )* */

    //上面注释掉，下面的逻辑是重新来梳理的
    val resultDf5Before = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_eo_bus_volpassrate") //
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()

    resultDf5Before.createOrReplaceTempView("hms_eo_bus_volpassrate");

    // 2021-03-29之前的
    /*     ss.sql("select project_name,bus_id,substr(stat_time,1,10) as stat_time,sum(1) bus_voltage_cnt,bus_voltage_status," +
         "sum(case when bus_voltage_status = 1 then 1 else 0 end) interval_upper," +
         "sum(case when bus_voltage_status = 2 then 1 else 0 end) interval_lower," +
         "sum(case when bus_voltage_status = 3 then 1 else 0 end) power_outage," +
         "sum(case when bus_voltage_status = 4 then 1 else 0 end) data_exception," +
         "max(bus_voltage) voltage_max," +
         "min(bus_voltage) voltage_min from hms_eo_bus_volpassrate a " +
         "  where stat_time >= '"+t1+"' and stat_time<= '"+t2+"'  group by project_name,bus_id,substr(stat_time,1,10),bus_voltage_status")
         .createOrReplaceTempView("t_etl3")*/
    /* 这里是汇总表把每分钟的数据汇总起来，求出五个状态的和（0-合格 1-越上限 2-越下限 3-停电 4-数据异常）
    *  2021-03-29改的 去掉了bus_voltage_status状态 和 sum(1) bus_voltage_cnt,bus_voltage_status
       加上：sum(case when bus_voltage_status = 0 then 1 else 0 end) bus_voltage_cnt
    * */
    ss.sql("select project_name,bus_id,substr(stat_time,1,10) as stat_time," +
      " sum(case when bus_voltage_status = 0 then 1 else 0 end) bus_voltage_cnt, " +
      "sum(case when bus_voltage_status = 1 then 1 else 0 end) interval_upper," +
      "sum(case when bus_voltage_status = 2 then 1 else 0 end) interval_lower," +
      "sum(case when bus_voltage_status = 3 then 1 else 0 end) power_outage," +
      "sum(case when bus_voltage_status = 4 then 1 else 0 end) data_exception," +
      "max(bus_voltage) voltage_max," +
      "min(bus_voltage) voltage_min from hms_eo_bus_volpassrate a " +
      "  where stat_time >= '" + t1 + "' and stat_time<= '" + t2 + "'  group by project_name,bus_id,substr(stat_time,1,10)")
      .createOrReplaceTempView("t_etl3")
    //ss.sql( "select * from t_etl3").show()


    //  2021-03-29 之前的逻辑
    /*
    //  2021-03-15 发现缺少power_outage,data_exception两个值 --以加
        val resultDf5 = ss.sql("select a.project_name,a.bus_id,a.stat_time,a.bus_voltage_cnt,a.bus_voltage_status,interval_upper,interval_lower,power_outage,data_exception,voltage_max,voltage_min,max(case when voltage_max=b.bus_voltage then b.stat_time end) time_max,max(case when voltage_min=b.bus_voltage then b.stat_time end) time_min,'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time  from t_etl3 a left join hms_eo_bus_volpassrate b on a.project_name=b.project_name and a.bus_id=b.bus_id and a.stat_time=substr(b.stat_time,1,10) group by a.project_name,a.bus_id,a.stat_time,a.bus_voltage_cnt,a.bus_voltage_status,interval_upper,interval_lower,power_outage,data_exception,voltage_max,voltage_min")
    */
    /*
    2021-03-15 发现缺少power_outage,data_exception两个值 --以加
    2021-03-29 改的：去掉 a.bus_voltage_status 因为上面逻辑已经去掉了 这里暂时不需要这个字段来做判断
    */
    val resultDf5 = ss.sql("select a.project_name,a.bus_id,a.stat_time,a.bus_voltage_cnt," +
      "interval_upper,interval_lower,power_outage,data_exception,voltage_max,voltage_min," +
      "max(case when voltage_max=b.bus_voltage then b.stat_time end) time_max," +
      "min(case when voltage_min=b.bus_voltage then b.stat_time end) time_min," +
      "'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time  from t_etl3 a " +
      "left join hms_eo_bus_volpassrate b on a.project_name=b.project_name " +
      "and a.bus_id=b.bus_id and a.stat_time=substr(b.stat_time,1,10) group by " +
      "a.project_name,a.bus_id,a.stat_time,a.bus_voltage_cnt,interval_upper,interval_lower,power_outage,data_exception,voltage_max,voltage_min ")

    //resultDf5.show();

    val tableName2021 = "hms_eo_bus_volpassrate_sum"
    MysqlDml.delete2(s"delete from $tableName2021 where stat_time between  substr('$startTime',1,10)  and  substr('$endTime',1,10)")
    //    resultDf5
    //写mysql
    resultDf5.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tableName2021)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()
    /////////////////2021/////

  }

  /**
   * 无功考核 增加工况 和机组功率因数 输出增加机组考核状态
   *
   * @param ss
   * @param sc
   * @param hTable
   * @param time
   * @param step
   */
  def calcReactivePower(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {
    import ss.implicits._
    //读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_eo_bus_volconf")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    jdbcDF.createOrReplaceTempView("hms_bus_voltage_scope_conf")
    JzCondition.calData(ss, sc, hTable, time, step)
    ss.sql("select project_name,case when substr(stat_quarter,6,1)=1 then concat(substr(stat_quarter,1,4),'/01/01 ')" +
      "when substr(stat_quarter,6,1)=2 then concat(substr(stat_quarter,1,4),'/04/01 ')" +
      "when substr(stat_quarter,6,1)=3 then concat(substr(stat_quarter,1,4),'/07/01 ')" +
      "when substr(stat_quarter,6,1)=4 then concat(substr(stat_quarter,1,4),'/10/01 ') end as beg_date," +
      "case when substr(stat_quarter,6,1)=1 then concat(substr(stat_quarter,1,4),'/03/31 ')" +
      "when substr(stat_quarter,6,1)=2 then concat(substr(stat_quarter,1,4),'/06/30 ')" +
      "when substr(stat_quarter,6,1)=3 then concat(substr(stat_quarter,1,4),'/09/30 ')" +
      "when substr(stat_quarter,6,1)=4 then concat(substr(stat_quarter,1,4),'/12/31 ') end as end_date," +
      "lpad(stat_beg_time,8,'0') as beg_hhmmss,lpad(stat_end_time,8,'0') as end_hhmmss,bus_voltage_upper,bus_voltage_lower,bus_voltage_deactive from hms_bus_voltage_scope_conf").createOrReplaceTempView("hms_bus_voltage_scope_conf2")
    //ss.sql("select * from hms_bus_voltage_scope_conf2").show()

    val jdbc2DF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url)
      .option("dbtable", "hms_bus_voltage_point_conf")
      .option("user", MysqlDml.user)
      .option("password", MysqlDml.password)
      .option("driver", MysqlDml.driver2)
      .load().cache()
    jdbc2DF.createOrReplaceTempView("hms_bus_voltage_point_conf")
    //jdbc2DF.show()
    val arrayString: Array[String] = jdbc2DF.map { x => x.getString(1) }.collect()
    //val arrayString = Array("JKXT_DJ_AI03825","JKXT_DJ_AI03871")
    println(arrayString.mkString("---"))

    //GetData.getBySecond(ss,sc,hTable,arrayString,time,step)
    GetData.getByWebApi(ss, sc, hTable, arrayString, time, step, 300)
    ss.catalog.cacheTable("t_ods_hbase_edos_tmp")

    // 1:越线
    val resultDf = ss.sql("select a.project_name,bus_id,b.createTime as stat_time,c.bus_voltage_upper,c.bus_voltage_lower,round(b.pvalue,2) bus_voltage,case when b.pvalue between bus_voltage_lower and bus_voltage_upper then 0 when b.pvalue> bus_voltage_upper  then 1 else 2 end bus_voltage_status" +
      " from hms_bus_voltage_point_conf a" +
      " left join t_ods_hbase_edos_tmp b on a.point_code=b.checkPoint " +
      " left join hms_bus_voltage_scope_conf2 c on a.project_name=c.project_name and substr(b.createTime,1,10) between c.beg_date and c.end_date and  substr(b.createTime,12,8)>= c.beg_hhmmss and substr(b.createTime,12,8)<=c.end_hhmmss" +
      " where b.pvalue>c.bus_voltage_deactive" +
      " and substr(a.bus_id,-1)='母' ")
    //resultDf.orderBy("stat_time")show(200)
    resultDf.createOrReplaceTempView("t_result")
    val y = -20
    //按机组计算有功无功及工况

    //ss.sql("select substr(a.bus_id,-2),substr(a.bus_id,4,2) from hms_bus_voltage_point_conf a").show()
    val odsdf = ss.sql(s"select a.project_name,substr(a.bus_id,1,3) unit_name,b.createTime" +
      s",sum(case when substr(a.bus_id,-2)='无功' then b.pvalue end) power_no" +
      ",sum(case when substr(a.bus_id,-2)='有功' then b.pvalue end) power" +
      "  from hms_bus_voltage_point_conf a left join t_ods_hbase_edos_tmp b on a.point_code=b.checkPoint " +
      " where substr(a.bus_id,-1)='功' group by a.project_name,substr(a.bus_id,1,3),b.createTime ").createOrReplaceTempView("t_ods_tmp1")

    ss.sql("select project_name,unit_name,a.createTime,power_no,power,working_name" +
      ",power/sqrt(power_no*power_no+power*power) power_factor" +
      " from t_ods_tmp1 a left join t_jzgk_statu e on a.unit_name=e.crew_code and a.createTime=e.start_time").createOrReplaceTempView("t_ods_tmp2")


    val etldf = ss.sql("select project_name,stat_time" +
      ",(concat_ws(',',sort_array(collect_set(concat_ws(':',bus_id,bus_voltage))))) as voltage_unqualified_set" +
      ",max(bus_voltage_status) as bus_voltage_status" +
      " from t_result" +
      " group by project_name,stat_time")
    etldf.createOrReplaceTempView("t_etldf")
    val etldf2 = ss.sql(s"select a.project_name,a.stat_time,voltage_unqualified_set,bus_voltage_status" +
      s",sum(case when bus_voltage_status=1 and working_name='机组稳定运行态' and power_no>$y then 1 when bus_voltage_status=2 and working_name='机组稳定运行态' and (power_no<=0 or (power_no>0 and power_factor>0.9)) then 1 else 0 end) cnt" +
      s",str_to_map(concat_ws(',',sort_array(collect_set(concat_ws(':',unit_name,round(power_no,2)))))) reactive_unqualified_set" +
      s",concat_ws(',',sort_array(collect_set(concat_ws(':',unit_name,case when bus_voltage_status=1 and working_name='机组稳定运行态' and power_no>$y then 1 when bus_voltage_status=2 and working_name='机组稳定运行态' and (power_no<=0 or (power_no>0 and power_factor>0.9)) then 1 else 0 end)))) unit_unqualified_set" +
      "  from t_etldf a left join t_ods_tmp2 b on a.project_name=b.project_name and a.stat_time=b.createTime " +
      "  group by a.project_name,a.stat_time,voltage_unqualified_set,bus_voltage_status ")

    //ss.sql("select * from t_ods_tmp1 order by project_name,unit_name").show(500)
    //ss.sql("select * from t_ods_tmp2  order by project_name,unit_name").show(500)
    //etldf.show(500,0)
    //etldf2.show(500,0)


    etldf2.createOrReplaceTempView("t_etldf2")
    //regexp_replace(unit_unqualified_set,'..F:','')
    ss.sql("select project_name,stat_time,str_to_map(voltage_unqualified_set) voltage_unqualified_set" +
      ",bus_voltage_status,cnt,reactive_unqualified_set,str_to_map(unit_unqualified_set) unit_unqualified_set" +
      " from t_etldf2").createOrReplaceTempView("t_etldf3")
    //ss.sql("select * from t_etldf3 order by project_name ").show(50)
    //etldf.orderBy("stat_time").show(100,0)
    //etldf2.orderBy("createTime").show(100,0)
    val etldf3 = ss.sql("select a.project_name,stat_time,case when cnt>=1 then bus_voltage_status else 0 end status ," +
      "voltage_unqualified_set['一母'] bus_1,voltage_unqualified_set['二母'] bus_2,voltage_unqualified_set['三母'] bus_3,voltage_unqualified_set['四母'] bus_4," +
      "reactive_unqualified_set['01F'] reactive_power_01," +
      "reactive_unqualified_set['02F'] reactive_power_02," +
      "reactive_unqualified_set['03F'] reactive_power_03," +
      "reactive_unqualified_set['04F'] reactive_power_04," +
      "reactive_unqualified_set['05F'] reactive_power_05," +
      "reactive_unqualified_set['06F'] reactive_power_06," +
      "reactive_unqualified_set['07F'] reactive_power_07," +
      "reactive_unqualified_set['08F'] reactive_power_08," +
      "reactive_unqualified_set['09F'] reactive_power_09," +
      "reactive_unqualified_set['10F'] reactive_power_10," +
      "reactive_unqualified_set['11F'] reactive_power_11," +
      "reactive_unqualified_set['12F'] reactive_power_12," +
      "reactive_unqualified_set['13F'] reactive_power_13," +
      "reactive_unqualified_set['14F'] reactive_power_14," +
      "reactive_unqualified_set['15F'] reactive_power_15," +
      "reactive_unqualified_set['16F'] reactive_power_16," +
      "reactive_unqualified_set['17F'] reactive_power_17," +
      "reactive_unqualified_set['18F'] reactive_power_18," +
      "reactive_unqualified_set['19F'] reactive_power_19," +
      "reactive_unqualified_set['20F'] reactive_power_20," +
      "reactive_unqualified_set['21F'] reactive_power_21," +
      "cnt as unit_cnt," +
      "concat(nvl(unit_unqualified_set['01F'],'0'),nvl(unit_unqualified_set['02F'],'0'),nvl(unit_unqualified_set['03F'],'0'),nvl(unit_unqualified_set['04F'],'0'),nvl(unit_unqualified_set['05F'],'0'),nvl(unit_unqualified_set['06F'],'0'),nvl(unit_unqualified_set['07F'],'0'),nvl(unit_unqualified_set['08F'],'0'),nvl(unit_unqualified_set['09F'],'0'),nvl(unit_unqualified_set['10F'],'0'),nvl(unit_unqualified_set['11F'],'0'),nvl(unit_unqualified_set['12F'],'0'),nvl(unit_unqualified_set['13F'],'0'),nvl(unit_unqualified_set['14F'],'0'),nvl(unit_unqualified_set['15F'],'0'),nvl(unit_unqualified_set['16F'],'0'),nvl(unit_unqualified_set['17F'],'0'),nvl(unit_unqualified_set['18F'],'0'),nvl(unit_unqualified_set['19F'],'0'),nvl(unit_unqualified_set['20F'],'0'),nvl(unit_unqualified_set['21F'],'0')) unit_ass_type," +
      //"case when project_name='大江' then concat('0000000',regexp_replace(unit_unqualified_set,',','')) else concat(regexp_replace(unit_unqualified_set,',',''),'00000000000000') end unit_ass_type,"+
      "case when project_name='大江' then 2049*0.2 when project_name='二江' then 1090*0.2 end ass_quantity," +
      "'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time" +
      " from t_etldf3 a ")

    //etldf3.show()

    var startTime = time(0)
    val firstTime = time(0)
    val endTime = time(1)
    val tableName = "hms_eo_volpassrate"
    MysqlDml.delete2(s"delete from $tableName where stat_time between  '$startTime'  and  '$endTime'")

    //写mysql
    etldf3.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tableName)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()
    ss.catalog.uncacheTable("t_ods_hbase_edos_tmp")
  }
}
